﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Apache.NMS;
using Apache.NMS.ActiveMQ;
using Apache.NMS.Util;
using System.Xml;

namespace Parallel3M.parallel3m.framework
{
    public class MQParallelSource:ParallelProcessor
    {

        //MQ parameters
        public string connectionURI="activemq:tcp://localhost:61616";
        public IConnectionFactory NMSfactory;
        public IConnection NMSconnection;
        public ISession NMSsession;
        public IDestination NMSdestination;
        public string NMSQueue;
        public IMessageConsumer consumer;
        public bool isAsyn = false;       
        public string selector;
        /*
         *  (1) pcsd = PCS_ID
         *  (2) pcsd = "*" and domain in (domain condition of the selector)
         *  (3) other selector expanation condition
         */

        public MQParallelSource()
        {

        }

        public MQParallelSource(string uri, bool asyn)
        {
            connectionURI = uri;
            isAsyn = asyn;
            try
            {
                NMSfactory = new NMSConnectionFactory(connectionURI);
                NMSconnection = NMSfactory.CreateConnection();
                NMSsession = NMSconnection.CreateSession();
                
                NMSdestination = SessionUtil.GetDestination(NMSsession, NMSQueue);
                consumer = NMSsession.CreateConsumer(NMSdestination,selector);

                if (isAsyn)
                {
                    consumer.Listener += new MessageListener(onMessage);
                }
                    
                NMSconnection.Start();
                



            }
            catch
            {

            }
        }


        //asyn on message
        public void onMessage(IMessage receivedMsg)
        {
            ITextMessage message = (ITextMessage)receivedMsg;
            if (message != null)
            {
                payloadcount++;
                foreach (ParallelConnection outputConnection in outputConnections)
                {
                    outputConnection.Flowin(TransferMessage2Payload(message));
                }
            }
        }

        //get the payload from MQ server and put it into the queue
        public override void process()
        {
            if (!isAsyn)
            {
                foreach (ParallelConnection outputConnection in outputConnections)
                {
                    ParallelChannel channel = (ParallelChannel)outputConnection.to;
                    if (channel.queue.Count == 0)
                    {
                        ITextMessage message = consumer.Receive() as ITextMessage;
                        if (message != null)
                        {
                            payloadcount++;
                            outputConnection.Flowin(TransferMessage2Payload(message));
                        }
                    }
                }
            }            
        }

        public ParallelPayload TransferMessage2Payload(ITextMessage message)
        {
            ParallelPayload payload=new ParallelPayload();
            payload.id=(string)message.Properties[ParallelPayload.ID];
            payload.handlerid=(string)message.Properties[ParallelPayload.HANDLEID];
            payload.domain=(string)message.Properties[ParallelPayload.DOMAIN];
            payload.access=(string)message.Properties[ParallelPayload.ACCESS];
            payload.type=(string)message.Properties[ParallelPayload.TYPE];
            payload.routeTag=(string)message.Properties[ParallelPayload.ROUTETAG];
            payload.language=(string)message.Properties[ParallelPayload.LANGUAGE];
            payload.routeCount = (int)message.Properties[ParallelPayload.ROUTECOUNT];
            payload.pcs_source = (string)message.Properties[ParallelPayload.PCSSOURCE];
            payload.pcs_des = (string)message.Properties[ParallelPayload.PCSDESTINATION];

            if (payload.type == PAYLOADCONSTRANT.TYPE_REQUEST)
            {
                payload.request = message.Text;
            }
            else
            {
                payload.response = message.Text;
            }
            return payload;

        }

        //
        public override void PreParse()
        {
            base.PreParse();
            foreach (XmlNode childnode in node.ChildNodes)
            {
                if (childnode.Name == "url")
                {
                    connectionURI = childnode.InnerText;
                }
                else if (childnode.Name == "isasyn")
                {
                    string isasyn_str = childnode.InnerText;
                    if (isasyn_str == "true")
                    {
                        isAsyn = true;
                    }
                }
                else if (childnode.Name == "queneName")
                {
                    NMSQueue = childnode.InnerText;
                }
                else if (childnode.Name == "selector")
                {
                    selector = childnode.InnerText;
                    if(selector.Contains("{pcsid}")){
                        selector=selector.Replace("{pcsid}",application.pcsid);
                    }
                }

            }           

            try
            {
               NMSfactory = new NMSConnectionFactory(connectionURI);
               NMSconnection = NMSfactory.CreateConnection();
               NMSsession = NMSconnection.CreateSession();
               NMSdestination = SessionUtil.GetDestination(NMSsession, NMSQueue);
               consumer = NMSsession.CreateConsumer(NMSdestination,selector);

                

                if (isAsyn)
                {
                    consumer.Listener += new MessageListener(onMessage);
                }

                NMSconnection.Start();
            

            }
            catch
            {

            }
        }


    }
}
